package com.wuliu.context;

import cn.hutool.json.JSONUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wuliu.annotation.MessageConsumer;
import com.wuliu.annotation.MessageListener;
import com.wuliu.common.constant.QueueConstants;
import com.wuliu.common.core.domain.Message;
import com.wuliu.common.core.redis.RedisCache;
import com.wuliu.system.domain.SysSmsRecord;
import com.wuliu.system.service.ISysSmsRecordService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;

import javax.annotation.Resource;

/**
 * 消费者
 *
 * @author realue
 * @date 2023-02-03
 */
@MessageConsumer
@Slf4j
public class RedisMqConsumer {
    @Resource
    private RedisCache redisCache;
    @Resource
    private ISysSmsRecordService sysSmsRecordService;

    /**
     * 消费发送短信队列
     *
     * @param record
     */
    @MessageListener(streamKey = QueueConstants.SEND_SMS_QUEUE_KEY, consumerGroup = QueueConstants.SEND_SMS_QUEUE_GROUP,
            consumerName = QueueConstants.SEND_SMS_CONSUMER_NAME, mode = MessageListener.Mode.STREAM)
    @SneakyThrows
    public void consumerSms(ObjectRecord<String, Object> record) {
        String stream = record.getStream();
        log.info("消息发送短信队列:{}", record.getValue());
        ObjectMapper objectMapper = new ObjectMapper();
        Object obj = objectMapper.readValue(record.getValue().toString(), Object.class);
        Message message = JSONUtil.toBean(JSONUtil.toJsonStr(obj), Message.class);
        SysSmsRecord sysSmsRecord = JSONUtil.toBean(message.getContent().toString(), SysSmsRecord.class);
        sysSmsRecordService.sendSms(sysSmsRecord);
        redisCache.ack(QueueConstants.SEND_SMS_QUEUE_GROUP, record);
        redisCache.delete(stream, record.getId().toString());
    }

//    @MessageListener(streamKey = "streamKey", consumerGroup = "consumerGroup", consumerName = "consumerName", mode = MessageListener.Mode.STREAM)
//    public void testStream1(ObjectRecord<Object, Object> message) {
//        log.info("testStream1 组 consumerGroup 名 consumerName 消费消息 ===> " + JSONUtil.toJsonStr(message));
//        // 手动 ack
//        try {
//            // 模拟消息消费业务处理耗时
//            Thread.sleep(1000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//        redisCache.ack("consumerGroup", message);
//    }
//
//    /**
//     * 同一个组，不同的 consumerName 都可以进行消息的消费，与上面 testStream() 为竞争关系，消息仅被其中一个消费
//     *
//     * @param message msg
//     */
//    @MessageListener(streamKey = "streamKey", consumerGroup = "consumerGroup", consumerName = "consumerName", mode = MessageListener.Mode.STREAM)
//    public void testStream2(ObjectRecord<Object, Object> message) {
//        log.info("testStream2 组 consumerGroup 名 consumerName2 消费消息 ===> " + JSONUtil.toJsonStr(message));
//        redisCache.ack("consumerGroup", message);
//    }

    /**
     * 消费 streamKey 中，没有进行 ack 的消息
     *
     * @param message message
     */
    @MessageListener(
            streamKey = QueueConstants.SEND_SMS_QUEUE_KEY,
            consumerGroup = QueueConstants.SEND_SMS_QUEUE_GROUP,
            consumerName = QueueConstants.SEND_SMS_CONSUMER_NAME,
            pending = true,
            mode = MessageListener.Mode.STREAM)
    public void testStreamPending(ObjectRecord<Object, Object> message) {
        log.info("testStreamPending 组 consumerGroup 名 consumerName 消费未 ack 消息 ===> " + JSONUtil.toJsonStr(message));
        redisCache.ack(QueueConstants.SEND_SMS_QUEUE_GROUP, message);
    }
}